//! # 流处理器模块
//! 
//! 提供各种流处理器和操作符的实现，包括：
//! - 映射处理器（Map）
//! - 过滤处理器（Filter）
//! - 聚合处理器（Aggregate）
//! - 窗口处理器（Window）
//! - 转换处理器（Transform）
//! - 批处理器（Batch）
//! - 路由处理器（Route）
//! - 连接处理器（Join）

use async_trait::async_trait;
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize};
use tokio::sync::{RwLock, Mutex};
use std::sync::Arc;
use futures::Stream;
use tokio_stream::StreamExt;

use crate::core::{StreamProcessor, TypeInfo};
use crate::error::{StreamResult, StreamError};
use rustcloud_bus::prelude::Message;
use crate::window::*;

/// 处理器统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProcessorStats {
    /// 处理的消息总数
    pub total_processed: u64,
    /// 成功处理的消息数
    pub successful: u64,
    /// 失败的消息数
    pub failed: u64,
    /// 平均处理时间（毫秒）
    pub avg_processing_time_ms: f64,
    /// 最后处理时间
    pub last_processed: Option<chrono::DateTime<chrono::Utc>>,
}

impl Default for ProcessorStats {
    fn default() -> Self {
        Self {
            total_processed: 0,
            successful: 0,
            failed: 0,
            avg_processing_time_ms: 0.0,
            last_processed: None,
        }
    }
}

/// 带统计信息的处理器基础结构
#[derive(Debug)]
pub struct StatefulProcessor<P> {
    processor: P,
    stats: Arc<RwLock<ProcessorStats>>,
}

impl<P> StatefulProcessor<P> {
    pub fn new(processor: P) -> Self {
        Self {
            processor,
            stats: Arc::new(RwLock::new(ProcessorStats::default())),
        }
    }

    pub async fn get_stats(&self) -> ProcessorStats {
        self.stats.read().await.clone()
    }

    async fn update_stats(&self, success: bool, processing_time: Duration) {
        let mut stats = self.stats.write().await;
        stats.total_processed += 1;
        if success {
            stats.successful += 1;
        } else {
            stats.failed += 1;
        }
        
        // 更新平均处理时间
        let new_time_ms = processing_time.as_millis() as f64;
        stats.avg_processing_time_ms = 
            (stats.avg_processing_time_ms * (stats.total_processed - 1) as f64 + new_time_ms) / stats.total_processed as f64;
        
        stats.last_processed = Some(chrono::Utc::now());
    }
}

#[async_trait]
impl<P, I, O> StreamProcessor<I, O> for StatefulProcessor<P>
where
    P: StreamProcessor<I, O> + Send + Sync,
    I: Send + 'static,
    O: Send + 'static,
{
    async fn process(&self, input: I) -> StreamResult<O> {
        let start = Instant::now();
        let result = self.processor.process(input).await;
        let processing_time = start.elapsed();
        
        self.update_stats(result.is_ok(), processing_time).await;
        result
    }

    fn name(&self) -> &str {
        self.processor.name()
    }

    fn input_type(&self) -> TypeInfo {
        self.processor.input_type()
    }

    fn output_type(&self) -> TypeInfo {
        self.processor.output_type()
    }
}

/// 映射处理器
#[derive(Debug)]
pub struct MapProcessor<F> {
    name: String,
    mapper: F,
}

impl<F> MapProcessor<F> 
where
    F: Fn(Message) -> Message + Send + Sync + std::fmt::Debug,
{
    pub fn new(name: impl Into<String>, mapper: F) -> Self {
        Self {
            name: name.into(),
            mapper,
        }
    }
    
    /// 使用默认名称创建映射处理器
    pub fn with_mapper(mapper: F) -> Self {
        Self::new("map", mapper)
    }
}

#[async_trait]
impl<F> StreamProcessor<Message, Message> for MapProcessor<F>
where
    F: Fn(Message) -> Message + Send + Sync + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<Message> {
        Ok((self.mapper)(input))
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }
}

/// 异步映射处理器
#[derive(Debug)]
pub struct AsyncMapProcessor<F, Fut> {
    name: String,
    mapper: F,
    _marker: std::marker::PhantomData<Fut>,
}

impl<F, Fut> AsyncMapProcessor<F, Fut>
where
    F: Fn(Message) -> Fut + Send + Sync,
    Fut: std::future::Future<Output = StreamResult<Message>> + Send,
{
    pub fn new(name: impl Into<String>, mapper: F) -> Self {
        Self {
            name: name.into(),
            mapper,
            _marker: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<F, Fut> StreamProcessor<Message, Message> for AsyncMapProcessor<F, Fut>
where
    F: Fn(Message) -> Fut + Send + Sync + std::fmt::Debug,
    Fut: std::future::Future<Output = StreamResult<Message>> + Send + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<Message> {
        (self.mapper)(input).await
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }
}

/// 过滤处理器
#[derive(Debug)]
pub struct FilterProcessor<F> {
    name: String,
    predicate: F,
}

impl<F> FilterProcessor<F>
where
    F: Fn(&Message) -> bool + Send + Sync + std::fmt::Debug,
{
    pub fn new(name: impl Into<String>, predicate: F) -> Self {
        Self {
            name: name.into(),
            predicate,
        }
    }
    
    /// 使用默认名称创建过滤处理器
    pub fn with_predicate(predicate: F) -> Self {
        Self::new("filter", predicate)
    }
}

#[async_trait]
impl<F> StreamProcessor<Message, Option<Message>> for FilterProcessor<F>
where
    F: Fn(&Message) -> bool + Send + Sync + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<Option<Message>> {
        if (self.predicate)(&input) {
            Ok(Some(input))
        } else {
            Ok(None)
        }
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Option<Message>>()
    }
}

/// 聚合处理器
#[derive(Debug)]
pub struct AggregateProcessor<K, V, A, F>
where
    K: Send + Sync + std::hash::Hash + Eq + Clone + std::fmt::Debug,
    V: Send + Sync + Clone + std::fmt::Debug,
    A: Send + Sync + Clone + std::fmt::Debug,
    F: Fn(&A, &V) -> A + Send + Sync + std::fmt::Debug,
{
    name: String,
    key_extractor: fn(&Message) -> StreamResult<K>,
    value_extractor: fn(&Message) -> StreamResult<V>,
    aggregator: F,
    initial_value: A,
    state: Arc<RwLock<HashMap<K, A>>>,
    window_config: Option<WindowConfig>,
}

impl<K, V, A, F> AggregateProcessor<K, V, A, F>
where
    K: Send + Sync + std::hash::Hash + Eq + Clone + std::fmt::Debug + 'static,
    V: Send + Sync + Clone + std::fmt::Debug + 'static,
    A: Send + Sync + Clone + std::fmt::Debug + 'static,
    F: Fn(&A, &V) -> A + Send + Sync + std::fmt::Debug,
{
    pub fn new(
        name: impl Into<String>,
        key_extractor: fn(&Message) -> StreamResult<K>,
        value_extractor: fn(&Message) -> StreamResult<V>,
        aggregator: F,
        initial_value: A,
    ) -> Self {
        Self {
            name: name.into(),
            key_extractor,
            value_extractor,
            aggregator,
            initial_value,
            state: Arc::new(RwLock::new(HashMap::new())),
            window_config: None,
        }
    }
    
    /// 设置窗口配置
    pub fn with_window(mut self, config: WindowConfig) -> Self {
        self.window_config = Some(config);
        self
    }
    
    /// 获取当前聚合状态
    pub async fn get_state(&self) -> HashMap<K, A> {
        self.state.read().await.clone()
    }
    
    /// 清空聚合状态
    pub async fn clear_state(&self) {
        self.state.write().await.clear();
    }
}

#[async_trait]
impl<K, V, A, F> StreamProcessor<Message, HashMap<K, A>> for AggregateProcessor<K, V, A, F>
where
    K: Send + Sync + std::hash::Hash + Eq + Clone + std::fmt::Debug + 'static,
    V: Send + Sync + Clone + std::fmt::Debug + 'static,
    A: Send + Sync + Clone + std::fmt::Debug + 'static,
    F: Fn(&A, &V) -> A + Send + Sync + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<HashMap<K, A>> {
        let key = (self.key_extractor)(&input)?;
        let value = (self.value_extractor)(&input)?;
        
        let mut state = self.state.write().await;
        let current = state.get(&key).cloned().unwrap_or_else(|| self.initial_value.clone());
        let new_value = (self.aggregator)(&current, &value);
        state.insert(key, new_value);
        
        Ok(state.clone())
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<HashMap<K, A>>()
    }
}

/// 批处理器
#[derive(Debug)]
pub struct BatchProcessor {
    name: String,
    batch_size: usize,
    timeout: Duration,
    buffer: Arc<Mutex<VecDeque<Message>>>,
    last_flush: Arc<Mutex<Instant>>,
}

impl BatchProcessor {
    pub fn new(
        name: impl Into<String>,
        batch_size: usize,
        timeout: Duration,
    ) -> Self {
        Self {
            name: name.into(),
            batch_size,
            timeout,
            buffer: Arc::new(Mutex::new(VecDeque::new())),
            last_flush: Arc::new(Mutex::new(Instant::now())),
        }
    }
    
    /// 强制刷新缓冲区
    pub async fn flush(&self) -> StreamResult<Vec<Message>> {
        let mut buffer = self.buffer.lock().await;
        let batch: Vec<Message> = buffer.drain(..).collect();
        *self.last_flush.lock().await = Instant::now();
        Ok(batch)
    }
    
    /// 检查是否应该刷新
    async fn should_flush(&self) -> bool {
        let buffer = self.buffer.lock().await;
        let last_flush = self.last_flush.lock().await;
        
        buffer.len() >= self.batch_size || last_flush.elapsed() >= self.timeout
    }
}

#[async_trait]
impl StreamProcessor<Message, Option<Vec<Message>>> for BatchProcessor {
    async fn process(&self, input: Message) -> StreamResult<Option<Vec<Message>>> {
        {
            let mut buffer = self.buffer.lock().await;
            buffer.push_back(input);
        }
        
        if self.should_flush().await {
            let batch = self.flush().await?;
            if !batch.is_empty() {
                Ok(Some(batch))
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        }
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Option<Vec<Message>>>()
    }
}

/// 路由处理器
#[derive(Debug)]
pub struct RouteProcessor<F> {
    name: String,
    router: F,
}

impl<F> RouteProcessor<F>
where
    F: Fn(&Message) -> String + Send + Sync + std::fmt::Debug,
{
    pub fn new(name: impl Into<String>, router: F) -> Self {
        Self {
            name: name.into(),
            router,
        }
    }
}

/// 路由结果
#[derive(Debug, Clone)]
pub struct RouteResult {
    pub route: String,
    pub message: Message,
}

#[async_trait]
impl<F> StreamProcessor<Message, RouteResult> for RouteProcessor<F>
where
    F: Fn(&Message) -> String + Send + Sync + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<RouteResult> {
        let route = (self.router)(&input);
        Ok(RouteResult {
            route,
            message: input,
        })
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<RouteResult>()
    }
}

/// 转换处理器
#[derive(Debug)]
pub struct TransformProcessor<T, F>
where
    T: Send + Sync + std::fmt::Debug + 'static,
    F: Fn(Message) -> StreamResult<T> + Send + Sync + std::fmt::Debug,
{
    name: String,
    transformer: F,
    _marker: std::marker::PhantomData<T>,
}

impl<T, F> TransformProcessor<T, F>
where
    T: Send + Sync + std::fmt::Debug + 'static,
    F: Fn(Message) -> StreamResult<T> + Send + Sync + std::fmt::Debug,
{
    pub fn new(name: impl Into<String>, transformer: F) -> Self {
        Self {
            name: name.into(),
            transformer,
            _marker: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<T, F> StreamProcessor<Message, T> for TransformProcessor<T, F>
where
    T: Send + Sync + std::fmt::Debug + 'static,
    F: Fn(Message) -> StreamResult<T> + Send + Sync + std::fmt::Debug,
{
    async fn process(&self, input: Message) -> StreamResult<T> {
        (self.transformer)(input)
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<T>()
    }
}

/// 处理器组合器
#[derive(Debug)]
pub struct ProcessorChain {
    name: String,
    processors: Vec<Box<dyn StreamProcessor<Message, Message>>>,
}

impl ProcessorChain {
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            processors: Vec::new(),
        }
    }
    
    /// 添加处理器到链中
    pub fn add_processor(mut self, processor: Box<dyn StreamProcessor<Message, Message>>) -> Self {
        self.processors.push(processor);
        self
    }
    
    /// 获取处理器数量
    pub fn len(&self) -> usize {
        self.processors.len()
    }
    
    /// 检查链是否为空
    pub fn is_empty(&self) -> bool {
        self.processors.is_empty()
    }
}

#[async_trait]
impl StreamProcessor<Message, Message> for ProcessorChain {
    async fn process(&self, mut input: Message) -> StreamResult<Message> {
        for processor in &self.processors {
            input = processor.process(input).await?;
        }
        Ok(input)
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Message>()
    }
}

/// 并行处理器
#[derive(Debug)]
pub struct ParallelProcessor {
    name: String,
    processor: Box<dyn StreamProcessor<Message, Message>>,
    parallelism: usize,
}

impl ParallelProcessor {
    pub fn new(
        name: impl Into<String>,
        processor: Box<dyn StreamProcessor<Message, Message>>,
        parallelism: usize,
    ) -> Self {
        Self {
            name: name.into(),
            processor,
            parallelism: parallelism.max(1),
        }
    }
}

#[async_trait]
impl StreamProcessor<Vec<Message>, Vec<Message>> for ParallelProcessor {
    async fn process(&self, input: Vec<Message>) -> StreamResult<Vec<Message>> {
        use futures::future::try_join_all;
        
        let chunks: Vec<_> = input.chunks(self.parallelism).map(|chunk| chunk.to_vec()).collect();
        let mut results = Vec::new();
        
        for chunk in chunks {
            let futures: Vec<_> = chunk.into_iter()
                .map(|msg| self.processor.process(msg))
                .collect();
            
            let chunk_results = try_join_all(futures).await?;
            results.extend(chunk_results);
        }
        
        Ok(results)
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn input_type(&self) -> TypeInfo {
        TypeInfo::new::<Vec<Message>>()
    }

    fn output_type(&self) -> TypeInfo {
        TypeInfo::new::<Vec<Message>>()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustcloud_bus::message::MessageBuilder;
    use serde_json::json;

    #[tokio::test]
    async fn test_map_processor() {
        let processor = MapProcessor::new("test_map", |mut msg: Message| {
            let mut payload = msg.payload().clone();
            payload["processed"] = json!(true);
            *msg.payload_mut() = payload;
            msg
        });

        let input = MessageBuilder::new()
            .topic("test")
            .payload(json!({"value": 42}))
            .build().unwrap();

        let result = processor.process(input).await.unwrap();
        assert_eq!(result.payload()["processed"], json!(true));
        assert_eq!(result.payload()["value"], json!(42));
    }

    #[tokio::test]
    async fn test_filter_processor() {
        let processor = FilterProcessor::new("test_filter", |msg: &Message| {
            msg.payload()["active"].as_bool().unwrap_or(false)
        });

        let active_msg = MessageBuilder::new()
            .topic("test")
            .payload(json!({"active": true}))
            .build().unwrap();

        let inactive_msg = MessageBuilder::new()
            .topic("test")
            .payload(json!({"active": false}))
            .build().unwrap();

        let active_result = processor.process(active_msg).await.unwrap();
        assert!(active_result.is_some());

        let inactive_result = processor.process(inactive_msg).await.unwrap();
        assert!(inactive_result.is_none());
    }

    #[tokio::test]
    async fn test_batch_processor() {
        let processor = BatchProcessor::new("test_batch", 3, Duration::from_millis(100));

        let msg1 = MessageBuilder::new().topic("test").payload(json!(1)).build().unwrap();
        let msg2 = MessageBuilder::new().topic("test").payload(json!(2)).build().unwrap();
        let msg3 = MessageBuilder::new().topic("test").payload(json!(3)).build().unwrap();

        let result1 = processor.process(msg1).await.unwrap();
        assert!(result1.is_none());

        let result2 = processor.process(msg2).await.unwrap();
        assert!(result2.is_none());

        let result3 = processor.process(msg3).await.unwrap();
        assert!(result3.is_some());
        
        let batch = result3.unwrap();
        assert_eq!(batch.len(), 3);
    }

    #[tokio::test]
    async fn test_stateful_processor() {
        let map_processor = MapProcessor::new("test", |msg| msg);
        let stateful_processor = StatefulProcessor::new(map_processor);

        let msg = MessageBuilder::new().topic("test").payload(json!({})).build().unwrap();
        
        let _ = stateful_processor.process(msg).await.unwrap();
        
        let stats = stateful_processor.get_stats().await;
        assert_eq!(stats.total_processed, 1);
        assert_eq!(stats.successful, 1);
        assert_eq!(stats.failed, 0);
    }

    #[tokio::test]
    async fn test_processor_chain() {
        let processor1 = Box::new(MapProcessor::new("step1", |mut msg: Message| {
            let mut payload = msg.payload().clone();
            payload["step1"] = json!(true);
            *msg.payload_mut() = payload;
            msg
        }));
        
        let processor2 = Box::new(MapProcessor::new("step2", |mut msg: Message| {
            let mut payload = msg.payload().clone();
            payload["step2"] = json!(true);
            *msg.payload_mut() = payload;
            msg
        }));

        let chain = ProcessorChain::new("test_chain")
            .add_processor(processor1)
            .add_processor(processor2);

        let input = MessageBuilder::new()
            .topic("test")
            .payload(json!({}))
            .build().unwrap();

        let result = chain.process(input).await.unwrap();
        assert_eq!(result.payload()["step1"], json!(true));
        assert_eq!(result.payload()["step2"], json!(true));
    }
}